package com.atguigu.day07;

import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;

// select * from A inner join B on A.key = B.key;
public class Example5 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Tuple2<String, String>> leftStream = env
                .fromElements(
                        Tuple2.of("key-1", "left-1"),
                        Tuple2.of("key-1", "left-2"),
                        Tuple2.of("key-2", "left-1")
                );

        DataStreamSource<Tuple2<String, String>> rightStream = env
                .fromElements(
                        Tuple2.of("key-1", "right-1"),
                        Tuple2.of("key-1", "right-2"),
                        Tuple2.of("key-2", "right-1")
                );

        leftStream
                .connect(rightStream)
                .process(new InnerJoin())
                .print();

        env.execute();
    }

    public static class InnerJoin extends CoProcessFunction<Tuple2<String, String>, Tuple2<String, String>, String> {
        private ArrayList<Tuple2<String, String>> left = new ArrayList<>();
        private ArrayList<Tuple2<String, String>> right = new ArrayList<>();

        @Override
        public void processElement1(Tuple2<String, String> in1, Context ctx, Collector<String> out) throws Exception {
            left.add(in1);

            for (Tuple2<String, String> e : right) {
                if (in1.f0.equals(e.f0)) {
                    out.collect(in1 + " => " + e);
                }
            }
        }

        @Override
        public void processElement2(Tuple2<String, String> in2, Context ctx, Collector<String> out) throws Exception {
            right.add(in2);

            for (Tuple2<String, String> e : left) {
                if (in2.f0.equals(e.f0)) {
                    out.collect(e + " => " + left);
                }
            }
        }
    }
}
